package com.gugee.ins.data.analysis.listener;

import com.gugee.ins.data.analysis.config.InsAnalysisConfig;
import com.gugee.ins.data.analysis.service.InsAnalysisBrandCategoryService;
import com.gugee.ins.data.analysis.service.InsAnalysisBrandService;
import com.gugee.ins.data.common.consumer.AbstractGugeeConsumerHandler;
import com.gugee.ins.data.common.msg.InsBloggerMessage;
import com.gugee.ins.data.common.msg.InsBrandMessage;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.kafka.support.Acknowledgment;
import org.springframework.stereotype.Component;

import java.util.List;


/**
 * InsAnalysisBrand Listener
 *
 * @author Huangxin
 */
@Component
@Slf4j
public class InsAnalysisBrandListener {

    /**
     * InsAnalysisBrand
     */
    @Autowired
    private InsAnalysisConfig insAnalysisConfig;

    /**
     * InsAnalysisBrand
     */
    @Autowired
    private InsAnalysisBrandService insAnalysisBrandService;

    /**
     * InsAnalysisBrandCategory
     */
    @Autowired
    private InsAnalysisBrandCategoryService insAnalysisBrandCategoryService;

    private InsAnalysisBrandHandler analysisHandler = new InsAnalysisBrandHandler();

    private InsAnalysisBrandCategoryHandler analysisCategoryHandler = new InsAnalysisBrandCategoryHandler();

    /**
     * 监听任务,完成播主品牌分析
     * @param messages
     * @param ack
     */
    @KafkaListener(id="${kafka.consumer.group.insAnalysisBrand.id}",
            topics = {"${kafka.topics.insAnalysisBrand.online}"},
            containerFactory = "factoryInsAnalysisBrandAck")
    public void analysisBrand(List<InsBloggerMessage> messages, Acknowledgment ack) {
        analysisHandler.handle("InsAnalysisBrand",messages, insAnalysisConfig.getAcceptInsAnalysisBrandMsgTimeMinRange(),ack);
    }

    /**
     * 监听任务,完成播主品牌分析分类
     * @param messages
     * @param ack
     */
    @KafkaListener(id="${kafka.consumer.group.insAnalysisBrandCategory.id}",
            topics = {"${kafka.topics.insAnalysisBrandCategory.online}"},
            containerFactory = "factoryInsAnalysisBrandCategoryAck")
    public void analysisBrandCategory(List<InsBrandMessage> messages, Acknowledgment ack) {
        analysisCategoryHandler.handle("InsAnalysisBrandCategory",messages, insAnalysisConfig.getAcceptInsAnalysisBrandCategoryMsgTimeMinRange(),ack);
    }

    class InsAnalysisBrandHandler extends AbstractGugeeConsumerHandler<InsBloggerMessage>{
        @Override
        public void dealMsg(List<InsBloggerMessage> messages) {
            insAnalysisBrandService.handleAnalysisBrand(messages);
        }
    }

    class InsAnalysisBrandCategoryHandler extends AbstractGugeeConsumerHandler<InsBrandMessage>{
        @Override
        public void dealMsg(List<InsBrandMessage> messages) {
            insAnalysisBrandCategoryService.handleAnalysisBrandCategory(messages);
        }
    }
}
